package com.yanggu.flink.datastream_api.wordcount

import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala._

object StreamWordCount {

  def main(args: Array[String]): Unit = {
    //1. 从命令行中获取host和port。例如--input hdfs://hadoop100:9870/flink/checkpoint
    val params = ParameterTool.fromArgs(args)
    val host = params.get("host")
    val port = params.getInt("port")

    //2. 创建流处理环境
    val environment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()

    //全局切断任务链
    //environment.disableOperatorChaining()

    val socketStream = environment.socketTextStream(host, port)

    //导入一些隐士转换, 主要是类型转换
    import org.apache.flink.api.scala._
    val dataStream = socketStream
      //自定义任务调度规则, slotSharingGroup设置算子独占slot, 如果没有设置, 后面所有的算子都共享一个slot
      .flatMap(_.split(" "))/*.slotSharingGroup("a")*/
      //自定义任务调度规则, disableChaining: 和前后算子chain都断开flatMap、filter、map都是独立的subtask
      .filter(_.nonEmpty)/*.disableChaining()*/
      //自定义任务调度规则, startNewChain: 表示开启新的operator chain, 在这里断开flatMap和filter合在一起, map成为新的operator chain的起点
      .map((_, 1))/*.startNewChain()*/
      .keyBy(_._1)
      .sum(1)

    //打印输出
    dataStream.print()

    //启动executor, 执行任务
    environment.execute("Socket Stream Word Count")

  }

}
